package cn.ipanel.bigdata.dw.ods

import cn.ipanel.bigdata.boot.date.Day
import cn.ipanel.bigdata.boot.logger.Logger
import cn.ipanel.bigdata.utils.Util.rowTo
import org.apache.commons.lang3.exception.ExceptionUtils
import org.apache.spark.sql.{Dataset, Row}

/**
 * @author lzz
 * @environment IntelliJ IDEA 2020.3.1
 * @projectName bigdata_shanxi
 * @date 2023/03/21 15:53
 * @description:
 */
protected[ods] object Report extends AbsOdsHive("t_ods_report") {

  final val F_SERVICE   : String = "f_service"    // 业务ID
  final val F_TIMESTAMP : String = "f_timestamp"  // 该行为产生的绝对时间

  override def createTable: AbsOdsTable = Table()

  override def emptyTable: Dataset[Table] = {
    import IMPLICITS._
    spark.createDataset(spark.sparkContext.emptyRDD[Table])
  }

  override def extraColumnNames: Seq[String] = Seq(F_SERVICE, F_TIMESTAMP)
  override def extraColumnBuild: String = {
    s""" $F_SERVICE string,
       | $F_TIMESTAMP bigint""".stripMargin
  }

  def readToTable(day: Day): Dataset[Table] = {
    import IMPLICITS._
    try {
      find(day).map(Table(_))
    } catch {
      case e: Exception =>
        Logger.E(
          s""" read day[${day.toDate}] $getDBName.$getTBName failed.
             | Because: ${ExceptionUtils.getStackTrace(e)}
             |""".stripMargin)
        emptyTable
    }
  }

  class Table(f_time: String = NONE,
              f_device_id: String = NONE,
              f_region: String = NONE,
              f_extra: Map[String, String] = EMPTY_MAP,
              f_date: Int = INT0,
              var f_service: String = NONE,
              var f_timestamp: Long = LONG0) extends AbsOdsTable(f_time, f_device_id, f_region, f_extra, f_date) {

    override def toParam: Array[Any] = Array(f_service, f_timestamp)

    override def _init_(row: Row): Table = {
      super._init_(row)
      f_service             = rowTo[String](row, F_SERVICE, NONE)
      f_timestamp           = rowTo[Long](row, F_TIMESTAMP, LONG0)
      this
    }
  }

  object Table {

    def apply(row: Row): Table = Table()._init_(row)
    def apply(): Table = new Table()
  }

}